import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object TopN {
  def main(args:Array[String]) : Unit = {
    val sparkConf = new SparkConf().setAppName("TopN").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val data = sc.textFile("data/access.log")

    val filterRDD: RDD[String] = data.filter(x=>x.split(" ").length>10)
    val urlAndOne = filterRDD.map(x=>x.split(" ")(10)).map((_,1))

    val result = urlAndOne.reduceByKey(_+_)
    val sortedRDD = result.sortBy(_._2,false)
    val top5 = sortedRDD.take(5)
    top5.foreach(println)
    sc.stop()
  }
}
